package net.bwie.dwd;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;


public class DwdEntranceLog {
    public static void main(String[] args) throws Exception {
        // 1.表执行环境
        TableEnvironment tableEnv = getTableEnv();

        // 2.输入表 读取数据
        createInputTable(tableEnv);

        // 3.数据处理
        Table resultTable = handle(tableEnv);
//        tableEnv.createTemporaryView("resultTable", resultTable);
//        TableResult tableResult = tableEnv.executeSql("SELECT * FROM resultTable");
//        tableResult.print();

        // 4.映射表 写入Kafka表中
        createOutputTable(tableEnv);

        // 5.保存数据
        saveToSink(tableEnv, resultTable);

    }
    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }

    private static void createInputTable(TableEnvironment stream) {

        // 1 门禁刷卡事实表 ods_entrance_guard_log_clean
        stream.executeSql(
                "CREATE TABLE ods_entrance_guard_log_clean(\n" +
                        "    id STRING NOT NULL,\n" +
                        "    openMode STRING,\n" +
                        "    openTime TIMESTAMP(3),\n" +
                        "    ownerId INT,\n" +
                        "    proc_time as PROCTIME(),\n" +
                        "    WATERMARK FOR openTime AS openTime - INTERVAL '5' DAYS\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-entrance-guard-log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods_entrance_guard_log_clean2',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
//        TableResult tableResult = stream.executeSql("SELECT * FROM ods_entrance_guard_log_clean");
//        tableResult.print();

        // 2 用户维表 ods_ower_clean
        stream.executeSql(
                "CREATE TABLE ods_ower_clean(\n" +
                        "    id INT,\n" +
                        "    roomId INT,\n" +
                        "    owerCode STRING,\n" +
                        "    owerName STRING,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerIdcard STRING,\n" +
                        "    owerTel STRING,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    remark STRING,\n" +
                        "    createTime TIMESTAMP(3),\n" +
                        "    operatorTime TIMESTAMP(3),\n" +
                        "    proc_time as PROCTIME(),\n" +
                        "    WATERMARK FOR createTime AS createTime - INTERVAL '5' DAYS\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris.ower',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );
//        stream.executeSql("SELECT * FROM ods_ower_clean").print();

        // 3 房间维表 ods_room_clean
        stream.executeSql(
                "CREATE TABLE ods_room_clean(\n" +
                        "    id INT,\n" +
                        "    houseId INT,\n" +
                        "    roomCode STRING,\n" +
                        "    roomName STRING,\n" +
                        "    peopleNum INT,\n" +
                        "    remark STRING,\n" +
                        "    proc_time as PROCTIME()\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris.room',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );

//        stream.executeSql("SELECT * FROM ods_room_clean").print();

        // 4 楼栋维表 house
        stream.executeSql(
                "CREATE TABLE ods_house (\n" +
                        "    id INT,\n" +
                        "    cid INT,\n" +
                        "    houseCode STRING,\n" +
                        "    houseName STRING,\n" +
                        "    buildStartTime STRING,\n" +
                        "    buildEndTime STRING,\n" +
                        "    houseArea STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris.house',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );
//        stream.executeSql("SELECT * FROM ods_house").print();

        // 5 社区维表 community
        stream.executeSql(
                "CREATE TABLE ods_community (\n" +
                        "    id INT,\n" +
                        "    communityName STRING,\n" +
                        "    district_id INT,\n" +
                        "    city_id INT,\n" +
                        "    communityLocation STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris.community',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );
//        stream.executeSql("SELECT * FROM ods_community").print();

        // 6 地区维表 district
        stream.executeSql(
                "CREATE TABLE ods_district (\n" +
                        "    id INT,\n" +
                        "    pid INT,\n" +
                        "    code STRING,\n" +
                        "    name STRING,\n" +
                        "    remark STRING,\n" +
                        "    create_time STRING,\n" +
                        "    update_time STRING,\n" +
                        "    status STRING,\n" +
                        "    level STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris.district',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );
//        stream.executeSql("SELECT * FROM ods_district").print();

    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        // todo 创建临时视图 然后写入Kafka中
        tableEnv.createTemporaryView("EntranceGuardLog", resultTable);
        tableEnv.executeSql(
                "insert into dwd_entrance_guard_log_kafka select * from EntranceGuardLog"
        );
    }

    private static void createOutputTable(TableEnvironment stream) {
        // todo 创建Kafka表
        stream.executeSql(
                "create table dwd_entrance_guard_log_kafka (\n" +
                        "    id STRING,\n" +
                        "    districtName STRING,\n" +
                        // 小区 名称
                        "    communityName STRING,\n" +
                        // 楼栋编号
                        "    houseCode STRING,\n" +
                        // 房间编号
                        "    roomCode STRING,\n" +
                        // 业主 id
                        "    ownerId INT,\n" +
                        "    openMode STRING,\n" +
                        "    openTime TIMESTAMP(3),\n" +
                        "    PRIMARY KEY (id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'dwd_entrance_guard_log_kafka',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json'\n" +
                        ")"
        );
    }

    private static Table handle(TableEnvironment stream) {
        // todo 查询
        Table searchLogTable = stream.sqlQuery(
                "SELECT\n" +
                        "    log.id,\n" +
                        " dist.name as districtName  ,\n"+
                        "  comm.communityName,\n"+
                        "  house.houseCode,\n"+
                        "  room.roomCode,\n"+
                        "  ower.id,\n"+
                        "    log.openMode as openMode,\n" +
                        "    log.openTime\n" +
                        "FROM ods_entrance_guard_log_clean log\n" +
                        "-- 关联用户维表\n" +
                        "LEFT JOIN ods_ower_clean ower\n" +
                        "ON log.ownerId = ower.id\n" +
                        "-- 关联房间维表\n" +
                        "LEFT JOIN ods_room_clean room\n" +
                        "ON ower.roomId = room.id\n" +
                        "-- 关联楼栋维表\n" +
                        "LEFT JOIN ods_house house\n" +
                        "ON room.houseId = house.id\n" +
                        "-- 关联社区维表，获取社区ID(cId)\n" +
                        "LEFT JOIN ods_community comm\n" +
                        "ON house.cid = comm.id\n" +
                        "-- 获取地区维表\n" +
                        "left join ods_district dist\n" +
                        "on comm.district_id = dist.id\n"+
                        "where dist.name IS NOT null"
        );
        return searchLogTable;
    }






}
